/*
* Copyright (C) 2020 ~ 2024 chaigec All Rights Reserved.
*
* Author:     chaigec <chaigec@163.com>
* Maintainer: chaigec <chaigec@163.com>

 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
*/

#include "MkRtspSession.h"
#include "MkRtspRequest.h"
#include "MkRtspSubSession.h"
#include "MkRtspServer.h"
#include "MkRtspResponse.h"
#include "MkRtspAuthenticator.h"
#include "MkRtspPublisherSource.h"
#include "MkRtp/MkRtpServerPool.h"
#include "MkRtp/MkRtpUdpServer.h"
#include "MkRtspHeader.h"
#include "MkRtspConumer.h"
#include "MkEvent/MkEvent.h"
#include "MkEvent/MkTimer.h"
#include "MkUtil/MkUtil.h"
#include "MkUtil/MkLog.h"

#define  DefaultSessionTimeOut   60
CMkRtspSession::CMkRtspSession(CMkEvent *pEvent, CMkTimer* pTimer, CMkRtspServer* pServer, CMkRtpUdpServer* pUdpServer, const MkMediaCallbacks& MediaCbs)
    : CMkTcpSession(pEvent, pTimer, pServer)
    , m_pAuthor(new CMkRtspAuthenticator)
    , m_TimeOut(60)
    , m_nMaxLen(Len2K)
    , m_nPacketOffset(0)
    , m_nPacketMaxLen(Len2K)
    , m_nOffset(0)
    , m_pRtspConsumer(nullptr)
    , m_bPlaying(FALSE)
    , m_pRtspPublishSource(nullptr)
    , m_bStreamTcp(FALSE)
    , m_MediaCbs(MediaCbs)
    , m_pUdpServer(pUdpServer)
{
    MkString Nonce;
    CMkUtil::CreateRandomString(Len16, Nonce);
    m_pAuthor->SetNonce(Nonce);
    m_pRecvBuffer = (Uint8*)malloc(m_nMaxLen);
    m_pPacketBuffer = (Uint8*)malloc(m_nPacketMaxLen);
    CMkUtil::CreateRandomString(Len16, m_PlaySession);
}

Uint32 CMkRtspSession::SendEsPacket(const MkEsPacket& EsPacket, const CMkVideoParam& VideoParam, const CMkAudioParam& AudioParam)
{
    MkUnorderedMap<Uint32, CMkRtspSubSession*>::iterator it = m_MapSubsession.find(EsPacket.CodecType);
    if (it == m_MapSubsession.end()) {
        //stream not set up;
        return NoneError;
    }
    m_nPacketOffset = 0;
    CMkRtpPacket RtpPacket;
    RtpPacket.SetSsrc(it->second->GetSsrc());
    RtpPacket.SetPayloadType(it->second->GetPayloadType());
    RtpPacket.SetTimeStamp(EsPacket.Pts * (MkDefaultClockRate / 1000));
    return SendRtp(it->second, RtpPacket, EsPacket);
}

Uint32 CMkRtspSession::SendRtp(CMkRtspSubSession* pSubSession, CMkRtpPacket& RtpPacket, const MkEsPacket& EsPacket)
{
    switch (pSubSession->GetCodec()) {
    case MkCodecH264:
        SendH264Rtp(pSubSession, RtpPacket, EsPacket);
        break;
    case MkCodecH265:
        SendH265Rtp(pSubSession, RtpPacket, EsPacket);
        break;
    case MkCodecAac:
        SendAacRtp(pSubSession, RtpPacket, EsPacket);
    default:
        break;
    }
    return NoneError;
}


Uint32 CMkRtspSession::SendH264Rtp(CMkRtspSubSession* pSubSession, CMkRtpPacket& RtpPacket, const MkEsPacket& EsPacket)
{
    MkListBuffer::const_iterator it = EsPacket.BufferList.GetBufferList().begin();
    for (; it != EsPacket.BufferList.GetBufferList().end(); it++) {
        //signal packet   
        if (MkMaxRtpLen > it->GetLength()) {
            RtpPacket.SetSequenceNumber(pSubSession->GetSeqNumber());
            RtpPacket.SetMarker(TRUE);
            RtpPacket.PacketHeader(m_pPacketBuffer + m_nPacketOffset, m_nPacketMaxLen - m_nPacketOffset);
            m_nPacketOffset += RtpPacket.GetHeaderLen();
            RtpPacket.AppentPayloadBufferList(*it);
            SendRtpPacket(pSubSession, RtpPacket);
            RtpPacket.ClearPayload();
        } else {
            Uint32 nOffset = 0;
            Uint8 NaluType = ((const Uint8*)it->GetBuffer())[0];
            Uint8* pBegin = nullptr;
            nOffset++;
            while (nOffset < it->GetLength()) {
                RtpPacket.SetSequenceNumber(pSubSession->GetSeqNumber());
                pBegin = m_pPacketBuffer + m_nPacketOffset;
                if (1 == nOffset) {
                    //first packet
                    RtpPacket.SetMarker(FALSE);
                    pBegin[0] = MkH264PacketFuA | (NaluType & 0xE0);
                    pBegin[1] = 0x80 | (NaluType & 0x0F);
                    RtpPacket.AppentPayloadBufferList(pBegin, 2);
                    m_nPacketOffset += 2;
                    RtpPacket.AppentPayloadBufferList((const Uint8*)it->GetBuffer() + nOffset, MkMaxRtpLen);
                    nOffset += MkMaxRtpLen;
                } else if (nOffset + MkMaxRtpLen >= it->GetLength()) {
                    //end packet
                    pBegin[0] = MkH264PacketFuA | (NaluType & 0xE0);
                    pBegin[1] = 0x40 | (NaluType & 0x0F);
                    RtpPacket.SetMarker(TRUE);
                    RtpPacket.AppentPayloadBufferList(pBegin, 2);
                    m_nPacketOffset += 2;
                    RtpPacket.AppentPayloadBufferList((const Uint8*)it->GetBuffer() + nOffset, it->GetLength() - nOffset);
                    nOffset = it->GetLength();
                } else {
                    //middle packet
                    RtpPacket.SetMarker(FALSE);
                    pBegin[0] = MkH264PacketFuA | (NaluType & 0xE0);
                    pBegin[1] = 0x00 | (NaluType & 0x0F);
                    RtpPacket.AppentPayloadBufferList(pBegin, 2);
                    m_nPacketOffset += 2;
                    RtpPacket.AppentPayloadBufferList((const Uint8*)it->GetBuffer() + nOffset, MkMaxRtpLen);
                    nOffset += MkMaxRtpLen;
                }
                RtpPacket.PacketHeader(m_pPacketBuffer + m_nPacketOffset, m_nPacketMaxLen - m_nPacketOffset);
                m_nPacketOffset += RtpPacket.GetHeaderLen();
                SendRtpPacket(pSubSession, RtpPacket);
                RtpPacket.ClearPayload();
            }
        }
    }
    return NoneError;
}

Uint32 CMkRtspSession::SendH265Rtp(CMkRtspSubSession* pSubSession, CMkRtpPacket& RtpPacket, const MkEsPacket& EsPacket)
{
    MkListBuffer::const_iterator it = EsPacket.BufferList.GetBufferList().begin();
    for (; it != EsPacket.BufferList.GetBufferList().end(); it++) {
        //signal packet   
        if (MkMaxRtpLen > it->GetLength()) {
            RtpPacket.SetSequenceNumber(pSubSession->GetSeqNumber());
            RtpPacket.SetMarker(TRUE);
            RtpPacket.PacketHeader(m_pPacketBuffer + m_nPacketOffset, m_nPacketMaxLen - m_nPacketOffset);
            m_nPacketOffset += RtpPacket.GetHeaderLen();
            RtpPacket.AppentPayloadBufferList(*it);
            SendRtpPacket(pSubSession, RtpPacket);
            RtpPacket.ClearPayload();
        } else { //fu
            Uint32 nOffset = 0;
            const Uint8 *pData = (const Uint8 *)it->GetBuffer();
            Uint8 NaluType = 0;
            CMkMediaUtil::Get265NaluType(*pData, NaluType);
            Uint8 LayerId = (pData[0] & 0x01) << 5 | ((pData[1] >> 3) & 0x1F);
            Uint8 Tid = pData[1] & 0x07;
            Uint8* pBegin = nullptr;
            nOffset += 2;
            while (nOffset < it->GetLength()) {
                RtpPacket.SetSequenceNumber(pSubSession->GetSeqNumber());
                pBegin = m_pPacketBuffer + m_nPacketOffset;
                pBegin[0] = ((MkHevcPacketPaci & 0x3F) << 1) | ((LayerId >> 5) & 0x01);
                pBegin[1] = ((LayerId & 0x1F) << 3) | (Tid & 0x07);
                if (2 == nOffset) {
                    //first packet
                    RtpPacket.SetMarker(FALSE);
                    pBegin[2] = 0x80 | (NaluType & 0x3F);
                    RtpPacket.AppentPayloadBufferList(pBegin, 3);
                    m_nPacketOffset += 3;
                    RtpPacket.AppentPayloadBufferList(pData + nOffset, MkMaxRtpLen);
                    nOffset += MkMaxRtpLen;
                } else if (nOffset + MkMaxRtpLen >= it->GetLength()) {
                    //end packet
                    pBegin[2] = 0x40 | (NaluType & 0x3F);
                    RtpPacket.SetMarker(TRUE);
                    RtpPacket.AppentPayloadBufferList(pBegin, 3);
                    m_nPacketOffset += 3;
                    RtpPacket.AppentPayloadBufferList(pData + nOffset, it->GetLength() - nOffset);
                    nOffset = it->GetLength();
                } else {
                    //middle packet
                    RtpPacket.SetMarker(FALSE);
                    pBegin[2] = 0x00 | (NaluType & 0x3F);
                    RtpPacket.AppentPayloadBufferList(pBegin, 3);
                    m_nPacketOffset += 3;
                    RtpPacket.AppentPayloadBufferList(pData + nOffset, MkMaxRtpLen);
                    nOffset += MkMaxRtpLen;
                }
                RtpPacket.PacketHeader(m_pPacketBuffer + m_nPacketOffset, m_nPacketMaxLen - m_nPacketOffset);
                m_nPacketOffset += RtpPacket.GetHeaderLen();
                SendRtpPacket(pSubSession, RtpPacket);
                RtpPacket.ClearPayload();
            }
        }
    }
    return NoneError;
}

Uint32 CMkRtspSession::SendAacRtp(CMkRtspSubSession* pSubSession, CMkRtpPacket& RtpPacket, const MkEsPacket& EsPacket)
{
    MkListBuffer::const_iterator it = EsPacket.BufferList.GetBufferList().begin();
    for (; it != EsPacket.BufferList.GetBufferList().end(); it++) {
        RtpPacket.SetSequenceNumber(pSubSession->GetSeqNumber());
        RtpPacket.SetMarker(TRUE);
        RtpPacket.PacketHeader(m_pPacketBuffer + m_nPacketOffset, m_nPacketMaxLen - m_nPacketOffset);
        m_nPacketOffset += RtpPacket.GetHeaderLen();
        RtpPacket.AppentPayloadBufferList(*it);
        SendRtpPacket(pSubSession, RtpPacket);
        RtpPacket.ClearPayload();
    }
    return NoneError;
}

Uint32 CMkRtspSession::SendRtpPacket(CMkRtspSubSession* pSubSession, const CMkRtpPacket& RtpPacket)
{
    Uint32 ErrorCode = NoneError;
    if (m_bStreamTcp) {
        Uint8 *pBegin = m_pPacketBuffer + m_nPacketOffset;
        Uint16 nSize = RtpPacket.GetPayloadLength() + RtpPacket.GetHeaderLen();
        pBegin[0] = 0x24;
        pBegin[1] = 0x00;
        pBegin[2] = (nSize >> 8) & 0xFF;
        pBegin[3] = nSize & 0xFF;
        CMkBufferList SendList;
        SendList.Append(pBegin, 4);
        SendList.Append(RtpPacket.GetHeaderData(), RtpPacket.GetHeaderLen());
        if (!RtpPacket.GetPayloadData()) {
            SendList.Append(RtpPacket.GetPayloadBufferList());
        } else {
            SendList.Append(RtpPacket.GetPayloadData(), RtpPacket.GetPayloadLength());
        }
        ErrorCode = Send(SendList);
    } else {
        ErrorCode = pSubSession->SendRtpPacket(RtpPacket);
    }
    return ErrorCode;
}

void CMkRtspSession::OnSubSessionSetupDelay(void *lParam)
{
    CMkRtspSubSession* pSession = (CMkRtspSubSession*)lParam;
    if (pSession) {

    }
}

BOOL CMkRtspSession::IsAllSubSessionSetup()
{
    MkUnorderedMap<Uint32, CMkRtspSubSession*>::iterator it = m_MapSubsession.begin();
    for (; it != m_MapSubsession.end(); it++) {
        if (it->second->IsNotSetUp()) {
            return FALSE;
        }
    }
    return TRUE;
}

Uint32 CMkRtspSession::OnRecv()
{
    Uint32 nToRead = m_nMaxLen - m_nOffset;
    if (0 == nToRead) {
        Close();
        MkWarningLog("not has enough memery to recv rtsp session buffer\n");
        return SocketClosed;
    }
    Uint32 ErrorCode = Recv(m_pRecvBuffer + m_nOffset, nToRead);
    MkCheckErrorLog(ErrorCode, "rtmp client recv failed\n");

    m_nOffset += nToRead;
    //m_pRecvBuffer[m_nOffset] = 0;
    //MkDebugLog("recv rtsp:%s\n", m_pRecvBuffer);
    Uint32 nUsed = 0;
    Uint32 nCurUsed = 0;
    while (nUsed < m_nOffset) {
        nCurUsed = 0;
        if (0x24 == m_pRecvBuffer[nUsed]) {  //magic byte is 0x24
            ErrorCode = ParseRtspFrame(m_pRecvBuffer + nUsed, m_nOffset - nUsed, nCurUsed);
        } else {
            ErrorCode = ParseRtspMessage(m_pRecvBuffer + nUsed, m_nOffset - nUsed, nCurUsed);
        }
        if (NotHasEnoughBuffer == ErrorCode) {
            break;
        } else if (NoneError != ErrorCode) {
            MkWarningLog("rtsp recv error msg:%s\n", m_pRecvBuffer + nUsed);
            Close();
            return SocketClosed;
        }
        if (CMkRtspUtil::GetInstance().DebugMessageContent()) {
            MkString strContent = MkString((const Int8*)m_pRecvBuffer + nUsed, nCurUsed);
            MkDebugLog("rtsp session recv:\n%s\n", strContent.c_str());
        }
        nUsed += nCurUsed;
    }
    memmove(m_pRecvBuffer, m_pRecvBuffer + nUsed, m_nOffset - nUsed);
    m_nOffset -= nUsed;
    return NoneError;
}

Uint32 CMkRtspSession::OnEventRemoved()
{
    if (m_MediaCbs.ConsumerClosedCb
        && m_pRtspConsumer) {
        m_pRtspConsumer->SessionRemoved();
        m_MediaCbs.ConsumerClosedCb(m_pRtspConsumer->GetParentSessionId(), MkMediaConsumeRtsp, m_pRtspConsumer->GetConsumerId());
    } else if (m_MediaCbs.SourceDeleteCb
        && m_pRtspPublishSource) {
        m_pRtspPublishSource->SessionRemoved();
        m_MediaCbs.SourceDeleteCb(m_pRtspPublishSource->GetSession());
    }
    return CMkTcpSession::OnEventRemoved();
}

Uint32 CMkRtspSession::Check(const CMkTime& NowTime)
{
    return NoneError;
}

Uint32 CMkRtspSession::OnRtcpPacket(const CMkRtcpPacket& RtcpPacket, const MkString& Host, Uint16 Port)
{
    return NoneError;
}

Uint32 CMkRtspSession::OnRtpPacket(const MkCodecType& Codec, const CMkRtpPacket& RtpPacket, const MkString& Host, Uint16 Port)
{
    if (!m_pRtspPublishSource) {
        return SystemError;
    }
    MkCodecType CurCodec = Codec;
    if (MkCodecUnknown == Codec) {
        MkUnorderedMap<Uint32, CMkRtspSubSession*>::iterator it = m_MapSubsession.begin();
        for (; it != m_MapSubsession.end(); it++) {
            if (RtpPacket.GetSsrc() == it->second->GetSsrc()) {
                CurCodec = (MkCodecType)it->first;
                break;
            }
        }
    }
    switch (CurCodec) {
    case MkCodecH264:
        m_pRtspPublishSource->ParseH264RtpPacket(RtpPacket);
        break;
    case MkCodecH265:
        m_pRtspPublishSource->ParseH265RtpPacket(RtpPacket);
        break;
    case MkCodecAac:
        m_pRtspPublishSource->ParseAacRtpPacket(RtpPacket);
        break;
    default:
        break;
    }
    return NoneError;
}

Uint32 CMkRtspSession::ParseRtspFrame(Uint8* pBuf, Uint32 nLen, OUT Uint32& nUsed)
{
    if (nLen < 4) {
        return NotHasEnoughBuffer;
    }
    Uint8 Magic = pBuf[0];
    Uint8 Channel = pBuf[1];
    Uint32 FrameLen = (pBuf[2] << 8) | pBuf[3];
    if (FrameLen + 4 > nLen) {
        return NotHasEnoughBuffer;
    }
    nUsed = 4 + FrameLen;
    CMkRtpPacket RtpPacket;
    Uint32 ErrorCode = RtpPacket.Parse(pBuf + 4, FrameLen);
    MkCheckErrorLog(ErrorCode, "parse rtp packet failed\n");
    OnRtpPacket(MkCodecUnknown, RtpPacket, GetRemoteHost(), GetRemotePort());
    return NoneError;
}

Uint32 CMkRtspSession::ParseRtspMessage(Uint8* pBuf, Uint32 nLen, OUT Uint32& nUsed)
{
    CMkRtspRequest Req;
    Uint32 ErrorCode = Req.Parse((Int8*)pBuf, nLen, nUsed);
    MkCheckErrorLog(ErrorCode, "parse rtsp message failed\n");

    switch (Req.GetMethod()) {
    case MkRtspMethodOption:
        ErrorCode = OnOption(Req);
        break;
    case MkRtspMethodDescribe:
        ErrorCode = OnDescribe(Req);
        break;
    case MkRtspMethodSetup:
        ErrorCode = OnSetUp(Req);
        break;
    case MkRtspMethodPlay:
        ErrorCode = OnPlay(Req);
        break;
    case MkRtspMethodAnnounce:
        ErrorCode = OnAnnounce(Req);
        break;
    case MkRtspMethodTearDown:
        ErrorCode = OnTearDown(Req);
        break;
    case MkRtspMethodGetParameter:
        ErrorCode = OnGetParameter(Req);
        break;
    case MkRtspMethodPause:
        ErrorCode = OnPause(Req);
        break;
    case MkRtspMethodRecord:
        ErrorCode = OnRecord(Req);
        break;
    case MkRtspMethodSetParameter:
        ErrorCode = OnSetParameter(Req);
        break;
    default:
        ErrorCode = SendResponse(Req, MkRtspStateMethodNotAllowed);
        break;
    }
    if (NoneError != ErrorCode) {
        SendResponse(Req, MkRtspStateInternalServerError);
    }
    return NoneError;
}

Uint32 CMkRtspSession::OnOption(const CMkRtspRequest& Req)
{
    CMkRtspServer* pRtspServer = dynamic_cast<CMkRtspServer*>(GetParentServer());
    if (!pRtspServer) {
        return SystemError;
    }
    MkString Url = Req.GetUri();
    MkRtspStatusCode StatusCode = MkRtspStateOk;
    //MkString Path;
    //MkString Host;
    Uint16 Port;
    MkString UserName;
    MkString Password;
    CMkRtspResponse Res;
    //m_RequestUrl = Url;

    Uint32 ErrorCode = CMkRtspUtil::GetInstance().RtspUrlParse(Url, m_Host, Port, UserName, Password, m_Path);
    MkCheckErrorLog(ErrorCode, "url:%s parse failed\n", Url.c_str());

    MkMap<MkRtspHeaderType, MkString> MapHeader;
    MkString strSupportMethods;
    MkVector<MkRtspMethod>::const_iterator it = pRtspServer->GetSupportMethods().begin();
    for (; it != pRtspServer->GetSupportMethods().end(); it++) {
        strSupportMethods += CMkRtspUtil::GetInstance().GetRtspMethod(*it);
        if (it + 1 != pRtspServer->GetSupportMethods().end()) {
            strSupportMethods += ", ";
        }
    }
    MapHeader[MkRtspHeaderPublic] = strSupportMethods;
    return SendResponse(Req, MkRtspStateOk, MapHeader);

}

Uint32 CMkRtspSession::RemoveH264or5EmulationBytes(Uint8* pTo, Uint32 ToSize, const Uint8* pFrom, Uint32 FromSize)
{
    Uint32 nRetSize = 0;
    Uint32 Index = 0;
    while (Index < FromSize && nRetSize + 1 < ToSize) {
        if (Index + 2 < FromSize
            && pFrom[Index] == 0
            && pFrom[Index + 1] == 0
            && pFrom[Index + 2] == 3) {
            pTo[nRetSize] = pTo[nRetSize + 1] = 0;
            nRetSize += 2;
            Index += 3;
        } else {
            pTo[nRetSize] = pFrom[Index];
            nRetSize += 1;
            Index += 1;
        }
    }
    return nRetSize;
}

Uint32 CMkRtspSession::OnDescribe(const CMkRtspRequest& Req)
{
    if (!m_MediaCbs.GetSourceParamCb) {
        return InvalidParams;
    }
    CMkVideoParam VideoParam;
    CMkAudioParam AudioParam;
    Uint32 Duration;
    Uint32 FileSize;
    Uint32 ErrorCode = m_MediaCbs.GetSourceParamCb(m_Path, VideoParam, AudioParam, Duration, FileSize, m_SourceSession);
    if (NoneError != ErrorCode) {
        ErrorCode = SendResponse(Req, MkRtspStateNotFound);
        Close();
        return SocketClosed;
    }
    if (MkCodecUnknown != VideoParam.GetCodec()) {
        CMkSdpMedia SdpMedia;
        SdpMedia.SetMediaType("video");
        SdpMedia.SetProtocol("RTP/AVP");
        SdpMedia.SetPort(0);
        CMkSdpFmt SdpFmt;
        SdpFmt.SetPayload(96);
        SdpFmt.SetEncodingName(CMkMediaUtil::GetInstance().GetStringCodec(VideoParam.GetCodec()));
        SdpFmt.SetClockRate(MkDefaultClockRate);
        SdpFmt.SetSps(VideoParam.GetSps());
        SdpFmt.SetPps(VideoParam.GetPps());
        SdpFmt.SetVps(VideoParam.GetVps());
        if (MkCodecH264 == VideoParam.GetCodec()
            && VideoParam.GetSps().size() > 4) {
            const Uint8* pSps = (const Uint8*)VideoParam.GetSps().data();
            Uint32 SpsSize = VideoParam.GetSps().size();
            Uint8 *pSpsWeb = (Uint8*)malloc(SpsSize);
            unsigned vpsWEBSize = RemoveH264or5EmulationBytes(pSpsWeb, SpsSize, pSps, SpsSize);
            SdpFmt.SetH264PacketMode(1);
            SdpFmt.SetH264ProfileLevelId(pSpsWeb[1] << 16 | pSpsWeb[2] << 16 | pSpsWeb[3]);
            free(pSpsWeb);
        }
        if (MkCodecH265 == VideoParam.GetCodec()
            && VideoParam.GetVps().size() > 18) {
            const Uint8* pVps = (const Uint8*)VideoParam.GetVps().data();
            Uint32 VpsSize = VideoParam.GetVps().size();
            Uint8 *pVpsWeb = (Uint8*)malloc(VpsSize);
            unsigned vpsWEBSize = RemoveH264or5EmulationBytes(pVpsWeb, VpsSize, pVps, VpsSize);
            SdpFmt.SetH265ProfileSpace(pVpsWeb[6] >> 6);
            SdpFmt.SetH265ProfileId(pVpsWeb[6] & 0x1F);
            SdpFmt.SetH265LevelId(pVpsWeb[17]);
            SdpFmt.SetH265TierFlag((pVpsWeb[6] >> 5) & 0x01);
            Int8 pTmpBuf[Len16] = { 0 };
            snprintf(pTmpBuf, Len16, "%02X%02X%02X%02X%02X%02X",
                pVpsWeb[11], pVpsWeb[12], pVpsWeb[13],
                pVpsWeb[14], pVpsWeb[15], pVpsWeb[16]);
            SdpFmt.SetH265InteropConstraints(pTmpBuf);
            free(pVpsWeb);
        }
        SdpMedia.AddFmt(SdpFmt);
        SdpMedia.SetControlUri(Req.GetUri() + "/streamid=video");
        SdpMedia.SetVideoWidth(VideoParam.GetWidth());
        SdpMedia.SetVideoHeight(VideoParam.GetHeight());
        SdpMedia.SetVideoFrameRate(VideoParam.GetFrameRate());
        if (0 != VideoParam.GetBitRate()) {
            SdpMedia.SetBindWidth(CMkSdpBandWidth("AS", std::to_string(VideoParam.GetBitRate())));
        }
        CMkRtspSubSession* pSubSession = new CMkRtspSubSession(GetRemoteHost(), nullptr);
        pSubSession->SetSdpMedia(SdpMedia);
        m_MapSubsession[VideoParam.GetCodec()] = pSubSession;
    }
    if (MkCodecUnknown != AudioParam.GetCodec()) {
        CMkSdpMedia SdpMedia;
        SdpMedia.SetMediaType("audio");
        SdpMedia.SetProtocol("RTP/AVP");
        SdpMedia.SetPort(0);
        CMkSdpFmt SdpFmt;
        SdpFmt.SetPayload(97);
        SdpFmt.SetEncodingName(CMkMediaUtil::GetInstance().GetStringCodec(AudioParam.GetCodec()));
        SdpFmt.SetClockRate(CMkMediaUtil::AudioSampleRateIndexToRate(AudioParam.GetSampleRate()));
        SdpFmt.SetEncodingParams(AudioParam.GetChannel());
        SdpFmt.SetH264ProfileLevelId(1);
        SdpMedia.AddFmt(SdpFmt);
        SdpMedia.SetControlUri(Req.GetUri() + "/streamid=audio");
        CMkRtspSubSession* pSubSession = new CMkRtspSubSession(GetRemoteHost(), nullptr);
        pSubSession->SetSdpMedia(SdpMedia);
        m_MapSubsession[AudioParam.GetCodec()] = pSubSession;
    }

    CMkSdp Sdp;
    Sdp.SetVersion("0");
    Sdp.SetOrigin(CMkOrigin("-", "0", "0", "IN", "IP4", "127.0.0.1"));
    Sdp.SetSessionName("mk rtsp server");
    Sdp.SetConnection(CMkSdpConnection("IN", "IP4", m_Host));
    MkUnorderedMap<Uint32, CMkRtspSubSession*>::iterator itSub = m_MapSubsession.begin();
    for (; itSub != m_MapSubsession.end(); itSub++) {
        Sdp.AddMedia(itSub->second->GetSdpMedia());
    }
    MkString StrSdp;
    Sdp.ToString(StrSdp);

    CMkRtspResponse Res;
    MkMap<MkRtspHeaderType, MkString> MapHeader;
    MapHeader[MkRtspHeaderContentType] = "application/sdp";
    MapHeader[MkRtspHeaderContentBase] = Req.GetUri();
    MapHeader[MkRtspHeaderContentLength] = std::to_string(StrSdp.size());
    return SendResponse(Req, MkRtspStateOk, MapHeader, StrSdp);
}
Uint32 CMkRtspSession::OnSetUp(const CMkRtspRequest& Req)
{
    Uint32 ErrorCode = NoneError;
    MkString strSession = Req.GetHeader(MkRtspHeaderSession);
    if (!strSession.empty() && strSession != m_PlaySession) {
        SendResponse(Req, MkRtspStateSessionNotFound);
        return NoneError;
    }
    MkString Url = Req.GetUri();
    CMkRtspSubSession *pSubsession = nullptr;
    MkUnorderedMap<Uint32, CMkRtspSubSession*>::iterator it = m_MapSubsession.begin();
    for (; it != m_MapSubsession.end(); it++) {
        if (Url == it->second->GetUri()) {
            pSubsession = it->second;
            break;
        }
    }
    if (!pSubsession) {
        SendResponse(Req, MkRtspStateBadRequest);
        return NoneError;
    }
    pSubsession->SetTimeOut(DefaultSessionTimeOut);
    MkString StrTransport = Req.GetHeader(MkRtspHeaderTransport);
    if (m_PlaySession.empty()) {
        CMkUtil::CreateRandomString(Len16, m_PlaySession);
    }
    CMkRtspTransport Transport;
    Transport.ParseServerTransport(StrTransport);
    Uint32 Ssrc = CMkUtil::GetRandomNumber();
    MkRtpRemoter Remoter;
    Remoter.RtpPacketCb = nullptr;
    Remoter.RtcpPacketCb = [this](const CMkRtcpPacket& RtcpPacket, const MkString& Host, Uint16 Port) {
        return OnRtcpPacket(RtcpPacket, Host, Port);
    };
    Remoter.RtpRemoteHost = GetRemoteHost();
    Remoter.RtpRemotePort = Transport.GetClientRtpPort();
    Remoter.Payload = pSubsession->GetPayloadType();
    Remoter.RtpType = MkRtpSender;
    Remoter.Ssrc = Ssrc;

    if ("UDP" == Transport.GetLowTransport()) {
        pSubsession->SetRtpUdpServer(m_pUdpServer);
        pSubsession->GetRtpUdpServer()->AddRtpRemoter(Ssrc, Remoter);
        pSubsession->SetRtpRemotePort(Transport.GetClientRtpPort());
        Transport.SetServerRtpPort(pSubsession->GetRtpLocalUdpPort());
        Transport.SetServerRtcpPort(pSubsession->GetRtpLocalUdpPort() + 1);
    } else {
        m_bStreamTcp = TRUE;
    }
    pSubsession->SetSsrc(Ssrc);
    Transport.SetSsrc(Ssrc);
    Transport.SetMode("record" == Transport.GetMode() ? "" : "play");
    Transport.ToServerTransportString(StrTransport);
    MkMap<MkRtspHeaderType, MkString> MapHeaders;
    MapHeaders[MkRtspHeaderTransport] = StrTransport;
    MapHeaders[MkRtspHeaderSession] = m_PlaySession + ";timeout=" + std::to_string(DefaultSessionTimeOut);
    return SendResponse(Req, MkRtspStateOk, MapHeaders);
}
Uint32 CMkRtspSession::OnPlay(const CMkRtspRequest& Req)
{
    if (!IsAllSubSessionSetup()) {
        SendResponse(Req, MkRtspStateBadRequest);
        return NoneError;
    }
    MkString Session = Req.GetHeader(MkRtspHeaderSession);
    if (m_PlaySession.empty() || Session != m_PlaySession) {
        SendResponse(Req, MkRtspStateConferenceNotFound);
        return NoneError;
    }

    MkString strRange = Req.GetHeader(MkRtspHeaderRange);
    MkString strScale = Req.GetHeader(MkRtspHeaderScale);
    MkPlaybackControlType ControlType = MkControlUnknown;
    float fSpeed = 1.0f;
    Uint32 SeekTime = 0;
    if (!strScale.empty()) {
        ControlType = MkControlSetSpeed;
        fSpeed = std::stof(strScale);
    } else if (strRange.size() > 5) {
        strRange = strRange.substr(4, strRange.size() - 1);
        if (strRange == "now") {
            ControlType = MkControlRecovery;
        } else {
            SeekTime = std::stoi(strRange);
            ControlType = MkControlSeek;
        }
    }
    if (MkControlUnknown != ControlType
        && m_MediaCbs.ConsumerPlayCtrlCb
        && m_pRtspConsumer) {
        m_MediaCbs.ConsumerPlayCtrlCb(m_pRtspConsumer, ControlType, fSpeed, SeekTime);
    } else if (m_MediaCbs.ConsumerAddCb) {
        m_pRtspConsumer = new CMkRtspConsumer(this);
        m_MediaCbs.ConsumerAddCb(m_SourceSession, new CMkRtspConsume, m_pRtspConsumer);
    }
    Uint32 rtpTime = 0;
    MkString RtpInfo;
    MkUnorderedMap<Uint32, CMkRtspSubSession*>::iterator it = m_MapSubsession.begin();
    for (; it != m_MapSubsession.end(); it++) {
        if (!RtpInfo.empty()) {
            RtpInfo += ",";
        }
        RtpInfo += "url=" + it->second->GetUri() + ";seq=" + std::to_string(it->second->GetCurSeqNumber()) + ":rtptime=" + std::to_string(rtpTime);
    }
    MkMap<MkRtspHeaderType, MkString> MapHeaders;
    MapHeaders[MkRtspHeaderRtpInfo] = RtpInfo;
    MapHeaders[MkRtspHeaderSession] = m_PlaySession;
    Uint32 ErrorCode = SendResponse(Req, MkRtspStateOk, MapHeaders);
    MkCheckErrorLog(ErrorCode, "send play response failed\n");
    //return m_pSenderSource->Start();
    return ErrorCode;
}

Uint32 CMkRtspSession::OnAnnounce(const CMkRtspRequest& Req)
{
    CMkRtspServer *pRtspServer = dynamic_cast<CMkRtspServer*>(GetParentServer());
    if (!pRtspServer
        || Req.GetBody().empty()
        || "application/sdp" != Req.GetHeader(MkRtspHeaderContentType)
        || !m_MediaCbs.GetSourceParamCb) {
        SendResponse(Req, MkRtspStateBadRequest);
        return NoneError;
    }
    CMkVideoParam VideoParam;
    CMkAudioParam AudioParam;
    Uint32 Duration;
    Uint32 FileSize;
    Uint32 ErrorCode = m_MediaCbs.GetSourceParamCb(Req.GetUri(), VideoParam, AudioParam, Duration, FileSize, m_SourceSession);
    //find the source
    if (NoneError == ErrorCode) {
        SendResponse(Req, MkRtspStateBadRequest);//todo
        return NoneError;
    }
    MkDelete(m_pRtspPublishSource);
    m_pRtspPublishSource = new CMkRtspPublisherSource(this);
    m_pRtspPublishSource->StartMediaSource([this, pRtspServer](CMkMediaSource* pMediaSource, Uint32 ErrorCode) {
        if (NoneError == ErrorCode
            && m_MediaCbs.SourceAddCb) {
            m_MediaCbs.SourceAddCb(pMediaSource);
        }
        return NoneError;
    });
    CMkSdp Sdp;
    ErrorCode = Sdp.ParseContent(Req.GetBody());
    MkCheckErrorLog(ErrorCode, "parse announce sdp:%s failed\n", Req.GetBody().c_str());
    MkString RtpHost = Sdp.GetConnection().GetAddr();

    if (RtpHost.empty()
        || "0.0.0.0" == RtpHost
        || "127.0.0.1" == RtpHost) {
        RtpHost = Sdp.GetOrigin().GetAddr();
        if (RtpHost.empty()
            || "0.0.0.0" == RtpHost
            || "127.0.0.1" == RtpHost) {
            RtpHost = GetRemoteHost();
        }
    }

    MkVector<CMkSdpMedia>::const_iterator it = Sdp.GetMedia().begin();
    for (; it != Sdp.GetMedia().end(); it++) {
        CMkRtspSubSession *pSubSession = new CMkRtspSubSession(RtpHost, [this](const CMkVideoParam& VideoParam, const CMkAudioParam& AudioParam) {
            if (MkCodecUnknown != VideoParam.GetCodec()) {
                m_pRtspPublishSource->SetVideoParam(VideoParam);
            }
            if (MkCodecUnknown != AudioParam.GetCodec()) {
                m_pRtspPublishSource->SetAudioParam(AudioParam);
            }
        });
        if (!pSubSession) {
            return MallocFailed;
        }
        pSubSession->SetSdpMedia(*it);
        m_MapSubsession[pSubSession->GetCodec()] = pSubSession;
    }
    if (m_PlaySession.empty()) {
        CMkUtil::CreateRandomString(Len16, m_PlaySession);
    }
    if (m_pRtspPublishSource) {
        m_pRtspPublishSource->m_ResultCb(m_pRtspPublishSource, NoneError);
    }
    return SendResponse(Req, MkRtspStateOk);
}

Uint32 CMkRtspSession::OnTearDown(const CMkRtspRequest& Req)
{
    MkString Session = Req.GetHeader(MkRtspHeaderSession);
    if (m_PlaySession.empty() || Session != m_PlaySession) {
        SendResponse(Req, MkRtspStateConferenceNotFound);
        return NoneError;
    }
    if (m_pRtspConsumer&& m_MediaCbs.ConsumerClosedCb) {
        m_MediaCbs.ConsumerClosedCb(m_pRtspConsumer->GetParentSessionId(), MkMediaConsumeRtsp, m_pRtspConsumer->GetConsumerId());
    }
    return SendResponse(Req, MkRtspStateOk);
}

Uint32 CMkRtspSession::OnPause(const CMkRtspRequest& Req)
{
    if (m_pRtspConsumer
        && m_MediaCbs.ConsumerPlayCtrlCb) {
        m_MediaCbs.ConsumerPlayCtrlCb(m_pRtspConsumer, MkControlPause, 0.0f, 0);
    }
    return SendResponse(Req, MkRtspStateOk);
}
Uint32 CMkRtspSession::OnRecord(const CMkRtspRequest& Req)
{
    CMkRtspServer* pServer = dynamic_cast<CMkRtspServer*>(GetParentServer());
    if (!IsAllSubSessionSetup() || !pServer) {
        SendResponse(Req, MkRtspStateBadRequest);
        return NoneError;
    }
    MkString Session = Req.GetHeader(MkRtspHeaderSession);
    if (m_PlaySession.empty() || Session != m_PlaySession) {
        SendResponse(Req, MkRtspStateConferenceNotFound);
        return NoneError;
    }
    return SendResponse(Req, MkRtspStateOk);
}
Uint32 CMkRtspSession::OnSetParameter(const CMkRtspRequest& Req)
{
    return SendResponse(Req, MkRtspStateOk);
}
Uint32 CMkRtspSession::OnGetParameter(const CMkRtspRequest& Req)
{
    return SendResponse(Req, MkRtspStateOk);
}

Uint32 CMkRtspSession::SendResponse(const CMkRtspRequest& Req, const MkRtspStatusCode& RtspStatus, const MkMap<MkRtspHeaderType, MkString>& MapHeader/*=MkMap<MkRtspHeaderType, MkString>()*/, const MkString& RtspBody/*=""*/, const MkMapString& ExternHeader/* = MkMapString()*/)
{
    CMkRtspResponse Res;
    Res.AddHeader(MkRtspHeaderCSeq, std::to_string(Req.GetCSeq()));
    Res.AddHeader(MkRtspHeaderUserAgent, MkUserAgent);
    CMkTime NowTime;
    MkString GmtTime;
    NowTime.GetGMTString(GmtTime);
    Res.AddHeader(MkRtspHeaderDate, GmtTime);
    Res.SetStatusCode(RtspStatus);
    Res.SetExtensionHeader(ExternHeader);
    Res.AddBody(RtspBody);
    MkMap<MkRtspHeaderType, MkString>::const_iterator it = MapHeader.begin();
    for (; it != MapHeader.end(); it++) {
        Res.AddHeader(it->first, it->second);
    }
    if (!m_PlaySession.empty()) {
        Res.AddHeader(MkRtspHeaderSession, m_PlaySession);
    }
    MkString StrRes;
    Uint32 ErrorCode = Res.ToString(StrRes);
    MkCheckErrorLog(ErrorCode, "packet rtsp response failed\n");
    if (CMkRtspUtil::GetInstance().DebugMessageContent()) {
        MkDebugLog("rtsp session send response:\n%s\n", StrRes.c_str());
    }
    return  CMkTcpSession::Send((const Uint8*)StrRes.c_str(), StrRes.size());
}

CMkRtspSession::~CMkRtspSession()
{
    MkDelete(m_pAuthor);
    MkFree(m_pRecvBuffer);
    MkFree(m_pPacketBuffer);
    MkUnorderedMap<Uint32, CMkRtspSubSession*>::iterator it = m_MapSubsession.begin();
    for (; it != m_MapSubsession.end(); it++) {
        MkDelete(it->second);
    }
    m_MapSubsession.clear();
    MkDelete(m_pRtspPublishSource);
    MkDelete(m_pRtspConsumer);
}